package discovery

import (
	"context"
	"encoding/json"
	"fmt"
	"net/url"
	"path"
	"strconv"
	"strings"
	"time"

	"os"

	"go.etcd.io/etcd/clientv3"

	log "github.com/bilibili/kratos/pkg/log"
	mvccpb "github.com/coreos/etcd/mvcc/mvccpb"
	"google.golang.org/grpc/resolver"
)

// Config discovery configures.
type Config struct {
	Nodes  []string
	Region string
	Zone   string
	Env    string
	Host   string
	Prefix string
	Lease  int64
	DialTimeout time.Duration
}

// Discovery is discovery client.
type Discovery struct {
	c          *Config
	etcdClient *clientv3.Client
}

func fixConfig(c *Config) {
	if len(c.Nodes) == 0 {
		panic("conf nodes can not be nil")
	}
	if len(c.Prefix) == 0 {
		panic("conf prefix can not be empty")
	}
	if c.Lease <= 0 {
		c.Lease = 5
	}
	if c.DialTimeout == 0 {
		c.DialTimeout = time.Second
	}
	if c.Region == "" {
		c.Region = os.Getenv("REGION")
	}
	if c.Zone == "" {
		c.Zone = os.Getenv("ZONE")
	}
	if c.Env == "" {
		c.Env = os.Getenv("DEPLOY_ENV")
	}
	if c.Host == "" {
		c.Host, _ = os.Hostname()
	}
}

// New new a discovery client.
func New(c *Config) (d *Discovery) {
	fixConfig(c)
	d = &Discovery{
		c: c,
	}
	var err error
	d.etcdClient, err = clientv3.New(clientv3.Config{
		Endpoints: c.Nodes,
		DialTimeout: c.DialTimeout,
	})
	if err != nil {
		panic(err)
	}
	resolver.Register(d)
	return
}

func (d *Discovery) Register(ins *Instance) context.CancelFunc {
	ctx, cancel := context.WithTimeout(context.Background(),d.c.DialTimeout)
	key := path.Join(d.c.Prefix, ins.AppID, ins.Hostname)
	bs, _ := json.Marshal(ins)

	// lease
	lease := clientv3.NewLease(d.etcdClient)
	leaseResp, err := lease.Grant(ctx,d.c.Lease)
	if err != nil {
		cancel()
		log.Error("grpc: get lease of key(%s) err(%v)", key, err)
		panic(err)
	}
	leaseCh,err :=lease.KeepAlive(context.Background(), leaseResp.ID)
	if err != nil{
		log.Error("grpc: keepalive lease of key(%s) err(%v)", leaseResp.ID, err)
	}
	go func() {
		for {
			<-leaseCh
		}
	}()
	_, err = d.etcdClient.Put(context.Background(), key, string(bs), clientv3.WithLease(leaseResp.ID))
	if err != nil {
		cancel()
		log.Error("grpc: register etcd key(%s) err(%v)", key, err)
	} else {
		log.Info("grpc: register etcd key(%s) ok", key)
	}
	return cancel
}

// Scheme return discovery's scheme
func (d *Discovery) Scheme() string {
	return scheme
}

func (d *Discovery) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOption) (resolver.Resolver, error) {
	// etcd://prefix/service.nama/hostname
	dst := strings.SplitN(target.Endpoint, "://", 2)
	if len(dst) == 0 {
		return nil, fmt.Errorf("grpc: parse target.Endpoint(%s) failed! the endpoint is empty", target.Endpoint)
	}
	serverName := path.Join(target.Authority, dst[0])

	log.Info("%v", cc)
	r := &Resolver{
		d:  d,
		cc: cc,
	}
	go r.watcher(serverName)
	return r, nil
}

func (r *Resolver) watcher(serverName string) {
	// get
	inss := make(map[string]*Instance)
	resp, err := r.d.etcdClient.Get(context.Background(), serverName, clientv3.WithPrefix())
	if err != nil {
		log.Error("grpc: get serverName(%s) err(%v)", serverName, err)
	}
	for _, kv := range resp.Kvs {
		var ins Instance
		appID := string(kv.Key)
		err := json.Unmarshal(kv.Value, &ins)
		if err != nil {
			log.Error("grpc: unmarshal ins err(%v)", err)
		} else {
			inss[appID] = &ins
		}
	}
	if len(inss) == 0 {
		log.Info("grpc: get serverName(%s) is empty", serverName)
	} else {
		for key, val := range inss {
			log.Info("get inss:   %v      %+v", key, val)
		}
		r.newAddress(inss)
		log.Info("grpc: get serverName(%s) ok", serverName)
	}
	// watch
	wch := r.d.etcdClient.Watch(context.Background(), serverName, clientv3.WithPrefix())
	for {
		var ins Instance
		w_resp := <-wch
		for _, event := range w_resp.Events {
			appID := string(event.Kv.Key)
			if event.Type == mvccpb.PUT {
				err := json.Unmarshal(event.Kv.Value, &ins)
				if err != nil {
					log.Error("grpc: unmarshal ins err(%v)", err)
				} else {
					log.Info("grpc: watch key(%s) put", appID)
					inss[appID] = &ins
				}
			}
			if event.Type == mvccpb.DELETE {
				delete(inss, appID)
				log.Info("grpc: watch key(%s) delete", appID)
			}

			log.Info("watch inss:   %v      %+v", appID, inss[appID])
			if len(inss) > 0 {
				r.newAddress(inss)
			}
		}
	}
}

func (r *Resolver) newAddress(instances map[string]*Instance) {
	var (
		n           int
		totalWeight int64
		addrs       = make([]resolver.Address, 0, len(instances))
	)
	for _, ins := range instances {
		rpcAddr, color, weight := extractAddrs(ins)
		if rpcAddr == "" {
			log.Warn("grpc resolver: invalid rpc address(%s,%s,%v) found!", ins.AppID, ins.Hostname, ins.Addrs)
			continue
		}
		if weight <= 0 {
			if totalWeight == 0 {
				weight = 10
			} else {
				weight = totalWeight / int64(n)
			}
		}
		n++
		totalWeight += weight
		addr := resolver.Address{
			Addr:       rpcAddr,
			Type:       resolver.Backend,
			ServerName: ins.AppID,
			Metadata:   &MD{Weight: weight, Color: color},
		}
		addrs = append(addrs, addr)
	}
	r.cc.NewAddress(addrs)
}

func extractAddrs(ins *Instance) (addr, color string, weight int64) {
	color = ins.Metadata[MetaColor]
	weight, _ = strconv.ParseInt(ins.Metadata[MetaWeight], 10, 64)
	for _, a := range ins.Addrs {
		u, err := url.Parse(a)
		if err == nil && u.Scheme == serverScheme {
			addr = u.Host
		}
	}
	return
}

// Close stop all running process including discovery and register
func (d *Discovery) Close() error {
	return nil
}

type Resolver struct {
	d  *Discovery
	cc resolver.ClientConn
}

func (r *Resolver) ResolveNow(resolver.ResolveNowOption) {}
func (r *Resolver) Close()                               {}
